#include "threadpool.h"
#include <functional>
#include <iostream>
#include <thread>

const int TASK_MAX_THRESHHOLD = 1024;
const int THREAD_MAX_THRESHHOLD = 10;
const int THREAD_MAX_IDLE_TIME = 10;

ThreadPool::ThreadPool()
    : initThreadSize_(0)
    , curThreadSize_(0)
    , taskSize_(0)
    , taskQueMaxThreshHold_(TASK_MAX_THRESHHOLD)
    , poolMode_(PoolMode::MODE_FIXED)
    , isRunning_(false)
    , idleThreadSize_(0)
    , threadSizeThreshHold_(100)
{}

ThreadPool::~ThreadPool() 
{
    isRunning_ = false;
    // notEmpty_.notify_all();     // 通知阻塞在notEmpty_条件的线程(死锁位置)

    // 等待线程池中所有线程返回：1.阻塞中  2.执行任务中
    std::unique_lock<std::mutex> lock(taskQueMtx_);
    notEmpty_.notify_all();     // 通知阻塞在notEmpty_条件的线程
    exitCond_.wait(lock, [&]()->bool { return threads_.size() == 0; });
}

void ThreadPool::setMode(PoolMode mode)    // 设置线程池工作模式
{
    if(checkRunningState()) return;
    poolMode_ = mode;
}

void ThreadPool::setTaskQueMaxThreshHold(size_t threshhold)    // 设置任务队列阈值
{
    if(checkRunningState()) return;
    taskQueMaxThreshHold_ = threshhold;
}

void ThreadPool::setThreadSizeThreshHold(size_t threshhold)     // Cache模式上限阈值
{
    if(checkRunningState()) return;
    if(poolMode_ == PoolMode::MODE_CACHED) 
        threadSizeThreshHold_ = threshhold;
}


bool ThreadPool::checkRunningState() const 
{
    return isRunning_;
}

Result ThreadPool::submitTask(std::shared_ptr<Task> sp)  // 提交任务
{
    // 加锁保证push任务
    std::unique_lock<std::mutex> lock(taskQueMtx_);

    // 检查合法性，等待队列有空余
    // while (taskQue_.size() == taskQueMaxThreshHold_)
    //     notFull_.wait(lock);

    // 非notFull状态则阻塞，但不能一直阻塞，一定时间后队列仍满，则反馈提交失败
    if(!notFull_.wait_for(lock, std::chrono::seconds(1), 
        [&]()->bool { return taskQue_.size() < taskQueMaxThreshHold_; }))
    {
        std::cerr << "taskQueue is full, task submit fail\n";
        // return task->getResult();        // 线程处理完task，task就被析构了
        return Result(sp, false);      
    }
    
    // 有空余直接push
    taskQue_.emplace(sp);
    taskSize_++;

    // push task后，队列不为空，通知消费者线程
    notEmpty_.notify_all();

    // Cache模式下(小而快的任务)：根据任务量和空闲线程的数量，来决定是否创建更多的线程
    if(poolMode_ == PoolMode::MODE_CACHED &&
        taskSize_ > idleThreadSize_ && curThreadSize_ < threadSizeThreshHold_)
    {
        std::cout << "create new thread[" << std::this_thread::get_id() << "]\n";

        std::unique_ptr<Thread> ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
        // threads_.emplace_back(move(ptr));
        int threadId = ptr->getId();
        threads_.emplace(threadId, std::move(ptr));
        threads_[threadId]->start();    // 启动线程
        curThreadSize_++;
        idleThreadSize_++;
    }

    return Result(sp);
}

void ThreadPool::start(size_t initThreadSize)   // 启动线程池
{
    isRunning_ = true;      // 修改线程池状态

    initThreadSize_ = initThreadSize;
    curThreadSize_ = initThreadSize;
    
    // 创建线程
    for(int i = 0; i < initThreadSize_; ++i)
    {
        std::unique_ptr<Thread> ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
        // std::unique_ptr<Thread> ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this));
        // threads_.emplace_back(move(ptr));
        int threadId = ptr->getId();
        threads_.emplace(threadId, std::move(ptr));
    }

    // 启动线程
    for(int i = 0; i < initThreadSize_; ++i)
    {
        threads_[i]->start();
        idleThreadSize_++;
    }
}

void ThreadPool::threadFunc(int threadId)
{
    // std::cout << "begin: " << std::this_thread::get_id() << std::endl;
    // std::cout << "end: " << std::this_thread::get_id() << std::endl;

    auto lastTime = std::chrono::high_resolution_clock().now();

    // 线程不断的执行任务
    for(;;)
    {
        std::shared_ptr<Task> task;
        {
            // 加锁保证获取task时的线程安全
            std::unique_lock<std::mutex> lock(taskQueMtx_);

            std::cout << "尝试获取任务: " << std::this_thread::get_id() << std::endl;

            while (taskSize_ == 0)    // 锁 + 双重判断
            {
                // 没有任务且线程池析构时再回收每个线程
                if(!isRunning_)   
                {
                    threads_.erase(threadId);
                    exitCond_.notify_all();
                    std::cout << "threadId: [" << std::this_thread::get_id() << "] exit!\n";
                    return;
                }               

                // Cache模式下：如果线程空余超过60s，回收线程
                if(poolMode_ == PoolMode::MODE_CACHED)
                {
                    // 等待任务队列不为空时超时返回，每一秒钟返回一次 
                    if(std::cv_status::timeout == notEmpty_.wait_for(lock, std::chrono::seconds(1)))
                    {
                        auto now = std::chrono::high_resolution_clock().now();
                        auto dur = std::chrono::duration_cast<std::chrono::seconds>(now - lastTime);
                        if(dur.count() >= THREAD_MAX_IDLE_TIME &&
                            curThreadSize_ > initThreadSize_)
                        {
                            // 回收线程 -> 从容器中删除线程 -> 
                            // vector没办法在threadFunc()中定位具体的线程 --> unordered_map
                            threads_.erase(threadId);       // 根据id删线程
                            curThreadSize_--;
                            idleThreadSize_--;
                            std::cout << "threadId: [" << std::this_thread::get_id() << "] exit!\n";
                            return;
                        }
                    }
                }
                else notEmpty_.wait(lock);      // 没有任务时阻塞              

                // if(!isRunning_)     // 线程池回收
                // {
                //     threads_.erase(threadId);
                //     exitCond_.notify_all();
                //     std::cout << "threadId: [" << std::this_thread::get_id() << "] exit!\n";
                //     return;
                // }        
            }
   
            idleThreadSize_--;      // 线程被唤醒、空闲-1

            // 取任务
            task = taskQue_.front();
            taskQue_.pop();
            taskSize_--;

            std::cout << "获取任务成功: " << std::this_thread::get_id() << std::endl;

            // 若还有其它任务，通知其它线程
            if(taskQue_.size() > 0) notEmpty_.notify_all();
            // 取出一个任务后，队列不为满，可以提交任务
            notFull_.notify_all();
        }
        // 执行任务
        if(task != nullptr)
        {
            // task->run();    
            task->exec();     // 把任务的返回值设置到Result中，再执行任务
        }
        idleThreadSize_++;
        lastTime = std::chrono::high_resolution_clock().now();  
    }
 
    // // 对于没有被阻塞的线程（在执行任务的线程）
    // threads_.erase(threadId); 
    // exitCond_.notify_all();
    // std::cout << "threadId: [" << std::this_thread::get_id() << "] exit!\n";
}


/* 
    Thread function implement
*/ 
int Thread::generateId_ = 0;

Thread::Thread(ThreadFunc func) 
    : func_(func)
    , threadId_(generateId_++)
{}
Thread::~Thread() {}

void Thread::start()
{
    std::thread t(func_, threadId_);   // 创建线程
    t.detach();     // 分离线程
} 

int Thread::getId() const
{
    return threadId_;
}

/*
    class Task function：
*/
void Task::exec()
{
    if(result_ != nullptr)
    {
        result_->setRetVal(run());      // 封装一层，将执行后的结果带回Result内
    }
}

void Task::setResult(Result* res)
{
    result_ = res;
}


/*
    class Result function： 
*/
Result::Result(std::shared_ptr<Task> task, bool isValid)
    : isValid_(isValid)
    , task_(task)
{
    task->setResult(this); 
}


void Result::setRetVal(Any any) 
{
    any_ = std::move(any);
    sem_.post();
}   

Any Result::get()
{
    if(!isValid_)
    {
        return "";
    }
    sem_.wait();  
    return std::move(any_);
}